package cn.getech.data.development.controller.real;

import cn.getech.data.development.constant.Constant;
import cn.getech.data.development.constant.RealTableType;
import cn.getech.data.development.constant.YesOrNoEnum;
import cn.getech.data.development.entity.ConfConnect;
import cn.getech.data.development.entity.TableFieldInfo;
import cn.getech.data.development.entity.flink.BdpRealTableInfo;
import cn.getech.data.development.model.dto.RealTimeTaskSaveDto;
import cn.getech.data.development.model.vo.RealTimeTaskVO;
import cn.getech.data.development.model.vo.TableFieldNameInfoVO;
import cn.getech.data.development.service.BdpRealTableInfoService;
import cn.getech.data.development.service.ConfConnectService;
import cn.getech.data.development.service.RealTimeTaskService;
import cn.getech.data.development.service.TableFieldInfoService;
import cn.getech.data.development.utils.kafka.LocalKafkaUtil;
import cn.getech.data.intelligence.common.utils.R;
import cn.getech.system.center.annotation.SysLog;
import cn.getech.system.center.constant.OperationModulesEnum;
import cn.getech.system.center.constant.OperationTypeEnum;
import cn.getech.system.center.constant.module.ModuleEnum;
import com.alibaba.excel.ExcelReader;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiResponse;
import io.swagger.annotations.ApiResponses;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.IOUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;

import java.util.ArrayList;
import java.util.List;

@Slf4j
@Api(tags = "实时任务")
@RestController
@RequestMapping("/bdp/realTime/task")
public class RealTimeTaskController {

    @Autowired
    private RealTimeTaskService realTimeTaskService;
    @Autowired
    private TableFieldInfoService tableFieldInfoService;
    @Autowired
    private BdpRealTableInfoService realTableInfoService;
    @Autowired
    private ConfConnectService confConnectService;
    /**
     * 获取数据仓库分区字段
     */
    @ApiOperation("获取数据仓库分区字段")
    @GetMapping("/getPartFields")  //  /data-development/bdp/realTime/task/getPartFields
    public R getFileds(Integer id,
                       @RequestParam(required = false,defaultValue = "1") Integer flag) {
        if (id == null) {
            return R.error("id 不能为空");
        }
        List<TableFieldInfo> tableFieldInfos = tableFieldInfoService.list(
                new QueryWrapper<TableFieldInfo>().eq("table_id", id).eq("is_partition", flag)
                        .orderByAsc("field_order", "is_partition"));
        List<TableFieldNameInfoVO> tableFieldInfoVOS = new ArrayList<>();
        for (TableFieldInfo tableFieldInfo : tableFieldInfos) {
            TableFieldNameInfoVO tableFieldNameInfoVO = new TableFieldNameInfoVO();
            tableFieldNameInfoVO.setId(tableFieldInfo.getId());
            tableFieldNameInfoVO.setFieldName(tableFieldInfo.getFieldName());
            tableFieldNameInfoVO.setFieldType(tableFieldInfo.getFieldType());
            tableFieldNameInfoVO.setFieldAlias(tableFieldInfo.getFieldAlias());
            tableFieldInfoVOS.add(tableFieldNameInfoVO);
        }
        return R.okWithData(tableFieldInfoVOS);
    }

    @ApiOperation("获取数据仓库分区字段")
    @GetMapping("/getAllFields")  //  /data-development/bdp/realTime/task/getAllFields
    public R getAllFields(Integer id) {
        if (id == null) {
            return R.error("id 不能为空");
        }
        List<TableFieldInfo> tableFieldInfos = tableFieldInfoService.list(
                new QueryWrapper<TableFieldInfo>().eq("table_id", id)
                        .orderByAsc("field_order", "is_partition"));
        return R.okWithData(tableFieldInfos);
    }

    @ApiOperation("保存任务")
    @PostMapping("/save")
    @SysLog(value="保存任务",type= OperationTypeEnum.ADDCODE,
            modulesType = OperationModulesEnum.JOBMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    public R saveOrUpdate(@RequestBody RealTimeTaskSaveDto dto) {
        if (Constant.RealTimeTaskType.FLINK_SQL_DAG.getCode().equals(dto.getTaskType())) {
            if (CollectionUtils.isEmpty(dto.getLineList())) {
                return R.error("请确认所有节点都已经连线");
            }
        }
        return realTimeTaskService.saveOrUpdate(dto);
    }

    @ApiOperation("锁定/解锁")
    @PostMapping("/lock")
    @SysLog(value="锁定/解锁",type= OperationTypeEnum.UPDATECODE,
            modulesType = OperationModulesEnum.JOBMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    public R lock(@RequestBody RealTimeTaskSaveDto dto) {
        if (dto.getMenuId() == null || dto.getLockState() == null) {
            return R.error("参数缺失！");
        }
        return realTimeTaskService.lockOrUnlock(dto);
    }

    @ApiOperation("获取任务详情")
    @GetMapping("/getInfo/{menuId}")
    @ApiResponses({@ApiResponse(code = 200, message = "Success", response = RealTimeTaskVO.class)})
    public R getInfo(@PathVariable("menuId") Integer menuId) {
        RealTimeTaskVO vo = realTimeTaskService.getByMenuId(menuId);
        return R.okWithData(vo);
    }

    @ApiOperation("运行任务")
    @GetMapping("/run/{menuId}")
    @SysLog(value="运行任务",type= OperationTypeEnum.RUNCODE,
            modulesType = OperationModulesEnum.JOBMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    public R run(@PathVariable("menuId") Integer menuId) throws Exception {
        return realTimeTaskService.run(menuId);
    }
    @Value("${spring.kafka.bootstrap-servers:}")
    private String kafkaUrl;
    @ApiOperation("上传数据")
    @PostMapping("/data/upload") //  /bdp/realTime/task/data/upload
    @SysLog(value="运行任务",type= OperationTypeEnum.UPLOADCODE,
            modulesType = OperationModulesEnum.JOBMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    public  R upload(@RequestParam(value = "file") MultipartFile file,
                        Integer menuId,
                        Integer tableId) {
        ExcelReader excelReader = null;
        if (menuId == null) {
            return  R.error("menuId不能为空");
        }
        if (tableId == null) {
            return  R.error("tableId不能为空");
        }
        if (file == null) {
            return  R.error("数据文件不能为空");
        }
        if (file.getOriginalFilename()!=null && !file.getOriginalFilename().endsWith(".txt")) {
            return  R.error("仅支持.txt文件上传");
        }
        try {
            BdpRealTableInfo table = realTableInfoService.getById(tableId);
            if (table == null) {
                return  R.error("实时表不存在");
            }
            String topic = "local_topic_" + menuId + table.getTopicName();
            String url = kafkaUrl;
            if (RealTableType.OUTER.getCode().equals(table.getTableType())) {
                ConfConnect kafka = confConnectService.selectOne(table.getResourceId());
                if (kafka != null) {
                    url = kafka.getKafkaUrl();
                } else {
                    return R.error("表{}的kafka地址不存在", table.getTableName());
                }
            }
            LocalKafkaUtil.createTopic(topic,
                    1, null, url);
            List<String> list = IOUtils.readLines(file.getInputStream(), "UTF-8");
            LocalKafkaUtil.produceData(list, topic, url);
        } catch (Exception e) {
            log.error("导入错误:", e);
            return  R.error("数据上传错误:{}",e.getMessage());
        }
        return  R.ok();
    }

    @ApiOperation("本地运行任务")
    @GetMapping("/local/run/{menuId}")// /bdp/realTime/task/local/run/{menuId}
    @SysLog(value="运行任务",type= OperationTypeEnum.RUNCODE,
            modulesType = OperationModulesEnum.JOBMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    public R localRun(@PathVariable("menuId") Integer menuId) throws Exception {
        return realTimeTaskService.localRun(menuId);
    }

    @ApiOperation("重跑任务")
    @GetMapping("/reRun/{menuId}")
    @SysLog(value="重跑任务",type= OperationTypeEnum.RUNCODE,
            modulesType = OperationModulesEnum.JOBMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    public R reRun(@PathVariable("menuId") Integer menuId) throws Exception {
        return realTimeTaskService.reRun(menuId);
    }

    @ApiOperation("停止任务")
    @GetMapping("/cencel/{menuId}")
    @SysLog(value="停止任务",type= OperationTypeEnum.STOPCODE,
            modulesType = OperationModulesEnum.JOBMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)

    public R cencel(@PathVariable("menuId") Integer menuId) throws Exception {
        return realTimeTaskService.cancel(menuId);
    }

    @ApiOperation("暂停任务")
    @GetMapping("/pause/{menuId}")///  /bdp/realTime/task/pause/124
    @SysLog(value="暂停任务",type= OperationTypeEnum.STOPCODE,
            modulesType = OperationModulesEnum.JOBMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    public R pause(@PathVariable("menuId") Integer menuId) throws Exception {
        return realTimeTaskService.pause(menuId);
    }

    @ApiOperation("sql预览")
    @GetMapping("/sql/preview/{menuId}")///data-development/bdp/realTime/task/sql/preview/417
    public R preview(@PathVariable("menuId") Integer menuId) throws Exception {
        if (menuId== null) {
            return R.error("任务id不能为空");
        }
        return R.okWithData(realTimeTaskService.wrapFlinkSql(menuId, YesOrNoEnum.NO.getValue()));
    }


    @ApiOperation("检测")
    @GetMapping("/check/{menuId}") // /data-development/bdp/realTime/task/check/{menuId}
    @SysLog(value="运行任务",type= OperationTypeEnum.CHECKCODE,
            modulesType = OperationModulesEnum.JOBMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    public R check(@PathVariable("menuId") Integer menuId) {
        if (menuId == null) {
            return R.error("任务id不能为空");
        }
        return realTimeTaskService.checkSource(menuId);
    }
}
